package com.boot2.core.dao.batch.insert;

import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import java.util.List;
import java.util.Map;

/**
 * @author zhangweilin
 * @Description: 批量提交接口, 用来获取包装执行对象
 * @date 2019/7/7
 */

public interface BatchCommitWare<T> {
    Log loger = LogFactory.getLog(BatchCommitWare.class);

    /**
     * 定时任务线程池
     */
//    ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(50);


//    /**
//     * 获取包装执行对象:定时刷新提交
//     *
//     * @return
//     */
//    default BatchCommitInvoker getScheduleBatchCommitInvoker() {
//        return () -> {
//            String key = getKey();
//            BatchCommit<T> batchCommit = getBatchCommit(key);
//            int batchMaxCommitSize = batchCommit.getBatchMaxCommitSize();
//            long intervalCommitTime = batchCommit.getIntervalCommitTime();
//            int[] count = new int[1];
//            BatchCommit<T>[] batchCommitArr = new BatchCommit[1];
//            batchCommitArr[0] = batchCommit;
//            executor.scheduleWithFixedDelay(() -> {
//                if (!batchCommitArr[0].isEmpty()) {
////                    List<T> objList = new ArrayList<>();
////                    T o = null;
////                    int size = batchMaxCommitSize;
////                    while ((o = batchCommitArr[0].poll()) != null) {
////                        --size;
////                        objList.add(o);
////                        if (size == 0) {
////                            break;
////                        }
////                    }
//                    List<T> objList = batchCommitArr[0].pollList(batchMaxCommitSize);
//                    if (CollectionUtils.isNotEmpty(objList)) {
//                        count[0]++;
//                        long time1 = System.currentTimeMillis();
//                        saveBatchList(objList);
//                        long time2 = System.currentTimeMillis();
//                        loger.info("数据【" + key + "】第【" + count[0] + "】次定时触发批量提交，本批次共【" + objList.size() + "】条，共耗时【" + (time2 - time1) + "ms】");
//                        objList.clear();
//                    }
//                }
//            }, 0, intervalCommitTime, TimeUnit.MILLISECONDS);
//        };
//    }


    /**
     * 获取队列一满就立刻提交的执行器
     *
     * @return
     */
    default BatchCommitInvoker getQueueFullOrScheduleBatchCommitInvoker() {
        return () -> {
            try {
                String key = getKey();

                BatchCommit<T> batchCommit;
                do {
                    //如果获取不到，一直等到人为初始化参数了为止
                    Thread.sleep(3000L);
                    batchCommit = getBatchCommitWithoutInit(key);
                } while (batchCommit == null);

                int count = 0;
                while (true) {
                    //当队列长度没达到指定的值时，此方法处于阻塞状态，一旦达到指定值，此方法立刻被执行完毕而往下走
                    batchCommit.getLock();
                    if (loger.isDebugEnabled()) {
                        loger.debug("数据【" + key + "】,getLock调用成功返回");
                    }
                    int batchMaxCommitSize = batchCommit.getBatchMaxCommitSize();
                    if (loger.isDebugEnabled()) {
                        loger.debug("数据【" + key + "】,获得batchMaxCommitSize为:"+batchMaxCommitSize);
                    }
                    List<T> objList = batchCommit.pollList(batchMaxCommitSize);
                    if (loger.isDebugEnabled()) {
                        loger.debug("数据【" + key + "】,获得objList，size:"+objList.size());
                    }
                    if (CollectionUtils.isNotEmpty(objList)) {
                        count++;
                        long time1 = System.currentTimeMillis();
                        if (loger.isDebugEnabled()) {
                            loger.debug("数据【" + key + "】,第【" + count + "】次,objList不为空，准备调用saveBatchList()方法 ");
                        }
                        saveBatchList(objList);
                        if (loger.isDebugEnabled()) {
                            loger.debug("数据【" + key + "】, saveBatchList()方法调用完毕 ");
                        }
                        long time2 = System.currentTimeMillis();
                        if (loger.isDebugEnabled()) {
                            loger.debug("数据【" + key + "】第【" + count + "】次批量提交，本批次共【" + objList.size() + "】条，共耗时【" + (time2 - time1) + "ms】");
                        }
                        objList.clear();
                    }
                }
            } catch (Throwable e) {
                loger.error("批量提交模块异常",e);
                e.printStackTrace();
            }
        };
    }

    default BatchCommit<T> getBatchCommit(String key) {
        Map<String, BatchCommit<?>> taskMapQueue = BatchCommitInvoker.taskMapQueue;
        BatchCommit<T> batchCommit = (BatchCommit<T>) taskMapQueue.get(key);
        if (null == batchCommit) {
            batchCommit = new BatchCommit<>(key);
            batchCommit.setBatchCommitDataMap(BatchCommitHandler.batchCommitDataMap);
            taskMapQueue.put(key, batchCommit);
        }
        return batchCommit;
    }

    /**
     * 只获取，不初始化
     * @param key
     * @return
     */
    default BatchCommit<T> getBatchCommitWithoutInit(String key) {
        Map<String, BatchCommit<?>> taskMapQueue = BatchCommitInvoker.taskMapQueue;
        BatchCommit<T> batchCommit = (BatchCommit<T>) taskMapQueue.get(key);
        return batchCommit;
    }

    /**
     * 预先提交
     * @param t
     */
    default void preCommit(T t){
        String key = getKey();
        BatchCommit<T> batchCommit = getBatchCommitWithoutInit(key);
        batchCommit.preCommit(t);
    }


    /**
     * 返回向队列中添加数据时使用的key
     *
     * @return
     */
    String getKey();

    /**
     * 批量入库
     *
     * @param objectList
     * @return
     */
    <S extends T> List<S> saveBatchList(List<T> objectList);
}
